package com.galeno.javabasic;

import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/3015:19
 */
public class day12 {
    public static void main(String[] args) {
        StreamExecutionEnvironment e = StreamExecutionEnvironment.getExecutionEnvironment();
        e.enableCheckpointing(10000);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(e);

        DataStreamSource<String> lines = e.socketTextStream("galeno01", 8888);
        SingleOutputStreamOperator<Tuple2<String, Integer>> tp = lines.map(x -> {
            String[] split = x.split(",");
            return Tuple2.of(split[0], Integer.parseInt(split[1]));
        }).returns(new TypeHint<Tuple2<String, Integer>>() {
        });
        tableEnvironment.createTemporaryView("wc",tp,$("word"),$("count"));
        TableResult tableResult = tableEnvironment.executeSql("select word,sum(count) counts from wc group by word");
       // DataStream<Tuple2<Boolean, Row>> res = tableEnv.toRetractStream(tableResult, Row.class);
        DataStream<Tuple2<Boolean, Row>> tuple2DataStream = tableEnvironment.toRetractStream((Table) tableResult, Row.class);


    }
}
